package com.partner.share.work.mq.consumer;


import com.partner.share.work.mq.construction.ActiveMQTopicConst;
import com.partner.share.work.websocket.server.WebsocketService;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTextMessage;

import javax.jms.*;

/**
 * @author pengyangyan
 * @version 1.0.0
 * @date 2019年09月29日 17:18:00
 */
@Slf4j
public class Consumer implements MessageListener {



    private WebsocketService websocketService = null;

    private Destination destination = null;

    private Connection connection = null;

    private Session session = null;

    private MessageConsumer consumer = null;

    //是否停止
    private boolean stop = false;

    /**
     * 初始化对象时，执行init函数
     * @param ws webSocket服务
     * @throws Exception 异常
     */
    public Consumer(WebsocketService ws) throws Exception {
        init();
        websocketService = ws;
    }

    private void init() throws JMSException {
        //ActiveProperties activeProperties = SpringContextHolder.getBean("activeProperties",ActiveProperties.class);
        // 连接工厂是用户创建连接的对象.
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory( "admin", "admin", "tcp://127.0.0.1:61616/");
        //创建连接对象
        connection = connectionFactory.createConnection();
        // 创建会话用于创建消息的生产者，消费者和消息。是生产和消费的一个单线程上下文。
        session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        //客户用来指定他生产消息的目标还有他消费消息的来源的对象. (主题类型类似广播)
        destination = session.createQueue(ActiveMQTopicConst.TOPIC_NAME_WEBSOCKET_SYSTEM_NOTICE);
        // 会话创建消息的生产者将消息发送到目的地
        consumer = session.createConsumer(destination);
    }

    public void receiveMessage() throws JMSException{
        connection.start();
        consumer.setMessageListener(this);
    }


    @Override
    public void onMessage(Message message) {
        try {
            ActiveMQTextMessage activeMQTextMessage = (ActiveMQTextMessage) message;
            log.info("{}监听器监听到的内容:{}",this,activeMQTextMessage.getText());
            websocketService.broadcast(activeMQTextMessage.getText());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     *  关闭连接
     * @throws JMSException 异常
     */
    public void close() throws JMSException {
        if (consumer!=null){
            consumer.close();
        }
        if (session!=null){
            session.close();
        }
        if (connection!=null){
            connection.close();
        }
        log.info("消费者已关闭!");
    }
}
